package chen.bupt.mapreduce.board.semantic;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import chen.bupt.constant.Constants;
import chen.bupt.util.HDFSFileUtils;

/**
 * 将原来计算的tfidf进行近义词合并
 * 
 * @author forrest
 * 
 */
public class SynonymsTFIDFJob {
	private static String inputBase = Constants.BOARD_TFIDF;
	private static String outputBase = "/user/chenlingpeng/smth/boards_tfidf_synonyms/";
	private static final String synonymsPath = "/user/chenlingpeng/smth/synonyms.txt";

	public static class SynonymsTFIDFMap extends
			Mapper<LongWritable, Text, Text, NullWritable> {
		private Map<String, String> synMap = new HashMap<String, String>();

		@Override
		protected void setup(Context context) throws IOException,
				InterruptedException {
			List<String> synonyms = HDFSFileUtils.readLines(synonymsPath);
			for (String s : synonyms) {
				synMap.put(s.split(",")[1], s.split(",")[0]);
			}
		}

		@Override
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			if (!HDFSFileUtils.isInputFile(context))
				return;
			String[] val = value.toString().split("\\|");
			Map<String, Double> tmpMap = new HashMap<String, Double>();
			if (val.length < 2)
				return;
			for (int i = 1; i < val.length; i++) {
				String[] tmp = val[i].split(";");
				if (synMap.containsKey(tmp[0])) {
					tmp[0] = synMap.get(tmp[0]);
				}
				Double d = tmpMap.get(tmp[0]);
				if (d == null) {
					tmpMap.put(tmp[0], Double.parseDouble(tmp[1]));
				} else {
					tmpMap.put(tmp[0], d + Double.parseDouble(tmp[1]));
				}
			}
			StringBuffer sb = new StringBuffer();
			String info[] = val[0].split(";");
			sb.append(info[1] + ";" + info[2] + ";" + info[3] + ";" + info[4]);
			Set<String> set = tmpMap.keySet();
			for (String k : set) {
				sb.append(";" + k + ":" + tmpMap.get(k));
			}
			context.write(new Text(sb.toString()), NullWritable.get());
		}
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception {
		Configuration confg = new Configuration();
		confg.set("fs.default.name", "hdfs://10.1.1.26:31054");
		FileSystem fs = FileSystem.get(confg);
		FileStatus[] fsDirs = fs.listStatus(new Path(inputBase));
		for (FileStatus fss : fsDirs) {
			if (!fss.isDir())
				continue;
			Path dirName = fss.getPath();
			String boardId = dirName.getName();
			if (boardId.startsWith("_"))
				continue;
			if (!boardId.startsWith(args[0]))
				continue;
			Configuration conf = new Configuration();
			Job job = new Job(conf, "SynonymsTFIDFJob_" + boardId);
			job.setJarByClass(SynonymsTFIDFJob.class);
			job.setMapperClass(SynonymsTFIDFMap.class);
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(NullWritable.class);
			job.setNumReduceTasks(1);
			Path input = new Path(inputBase + boardId);
			Path output = new Path(outputBase + boardId);
			HDFSFileUtils.deleteFile(output, conf);
			FileInputFormat.addInputPath(job, input);
			FileOutputFormat.setOutputPath(job, output);
			job.waitForCompletion(true);
		}
	}

}
